package cn.monkey.io.transport.mq.redisson;

import cn.monkey.io.transport.core.FluxReceive;
import org.redisson.api.RFuture;
import org.redisson.api.RTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;

public final class RedissonFluxReceive extends FluxReceive {

    private static final Logger log = LoggerFactory.getLogger(RedissonFluxReceive.class);
    final List<Map.Entry<RTopic, RFuture<Integer>>> topicListenerIdEntries;

    public RedissonFluxReceive(List<RTopic> topicList) {
        this.topicListenerIdEntries = topicList.stream()
                .map(topic -> Map.entry(topic, topic.addListenerAsync(Object.class, (channel, msg) -> onInboundNext(msg))))
                .toList();
    }

    @Override
    public void dispose() {
        this.topicListenerIdEntries.forEach(e -> {
            RTopic topic = e.getKey();
            RFuture<Integer> future = e.getValue();
            final List<String> channelNames = topic.getChannelNames();
            future.whenComplete((id, ex) -> {
                if (ex != null) {
                    log.error("topic: {} listener add error:\n", channelNames, ex);
                    return;
                }
                topic.removeListenerAsync(id).whenComplete((v, ex1) -> {
                    if (ex1 != null) {
                        log.error("topic: {} listener remove error:\n", channelNames, ex1);
                    }
                });
            });
        });
        super.dispose();
    }
}
